import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author lyf
 * @Classname Producer
 * @Description TODO
 * @Date 2021/9/6 22:02
 */
public class Producer {
    private final static Logger log = LoggerFactory.getLogger(Producer.class);

    @Test
    public void sync() throws Exception {
        DefaultMQProducer sync = new DefaultMQProducer("sync-group");
        sync.setNamesrvAddr("192.168.115.128:9876");
        sync.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("sync-base", "Tag1", new String("sync第" + i + "条消息").getBytes());
            SendResult result = sync.send(message);
            SendStatus sendStatus = result.getSendStatus();
            String msgId = result.getMsgId();
            int queueId = result.getMessageQueue().getQueueId();
            log.info("第{}消息发送状态：{}，消息ID，队列：{}", i, sendStatus, msgId, queueId);
        }
        sync.shutdown();
    }

    @Test
    public void async() throws Exception {
        DefaultMQProducer sync = new DefaultMQProducer("async-group");
        sync.setNamesrvAddr("192.168.115.128:9876");
        sync.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("async-base", "Tag2", new String("async第" + i + "条消息").getBytes());
            sync.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult result) {
                    SendStatus sendStatus = result.getSendStatus();
                    String msgId = result.getMsgId();
                    int queueId = result.getMessageQueue().getQueueId();
                    log.info("消息发送成功：{}，消息ID，队列：{}", sendStatus, msgId, queueId);
                }

                @Override
                public void onException(Throwable e) {
                    log.info("消息发送失败ex：{}", e);
                }
            });
            TimeUnit.SECONDS.sleep(1);
        }
        sync.shutdown();
    }

    @Test
    public void oneWay() throws Exception {
        DefaultMQProducer sync = new DefaultMQProducer("oneWay-group");
        sync.setNamesrvAddr("192.168.115.128:9876");
        sync.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("oneWay-base", "Tag3", new String("oneWay" + i + "条消息").getBytes());
            sync.sendOneway(message);
            TimeUnit.SECONDS.sleep(1);
        }
        sync.shutdown();
    }
}
